package com.yanggu.mapreduce.topN;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Comparator;
import java.util.PriorityQueue;

/**
 * @version V1.0
 * @author: YangGu
 * @date: 2020/2/8 21:24
 * @description:
 */
public class TopNMapper extends Mapper<LongWritable, Text, FlowBean, NullWritable> {

    /**
     * 需要注意的是这里不能使用TreeMap或者是TreeSet。因为当总流量相同的时候，会进行覆盖
     * 这里使用的是java提供的优先队列。需要自己传入比较器(默认按照升序排序)
     * 最小堆的堆顶的元素就是最小的元素。如果当堆的总个数大于10的时候，移除堆顶的元素，也就是最小的元素
     */
    private PriorityQueue<FlowBean> priorityQueue = new PriorityQueue<>(10, Comparator.comparingInt(FlowBean::getSumFlow));

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //13470253144 180	180	360

        //1. 获取一行数据
        String line = value.toString();

        //2. 切割字符串
        String[] fields = line.split("\t");

        //3. 封装数据
        FlowBean flowBean = new FlowBean();
        flowBean.setPhone(fields[0]);
        flowBean.setUpFlow(Integer.parseInt(fields[1]));
        flowBean.setDownFlow(Integer.parseInt(fields[2]));
        flowBean.setSumFlow(Integer.parseInt(fields[3]));

        //4. 如果小于10个，直接添加添加即可
        if (priorityQueue.size() < 10) {
            //添加到优先队列中
            priorityQueue.add(flowBean);
        } else {
            //如果大于等于10个需要与堆顶元素进行判断
            FlowBean minSumFlow = priorityQueue.peek();
            //如果新的bean对象的总流量大于堆顶bean对象(也就是堆中最小的总流量的对象)
            //这里不考虑相等的情况
            if (flowBean.getSumFlow() > minSumFlow.getSumFlow()) {
                //取出堆顶的元素
                priorityQueue.poll();
                //同时添加新的元素
                priorityQueue.add(flowBean);
            }
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        //在cleanup方法中实现写出。这里直接遍历即可。因为在归并排序的时候仍然会进行排序
        for (FlowBean flowBean : priorityQueue) {
            context.write(flowBean, NullWritable.get());
        }
    }
}
